Flink: Support Lookup Join using full in-memory lookup cache by Guosmilesmile · Pull Request #18144 · apache/iceberg

Guosmilesmile · 2026-09-16T16:47:19Z

This PR adds lookup join support for the Iceberg Flink table source, using a full in-memory lookup cache.

The implementation enables Iceberg tables to be used as temporal lookup join dimensions in Flink SQL.

Supported Features

Lookup join against Iceberg table source

Supports Flink SQL temporal lookup join syntax:

LEFT JOIN iceberg_catalog.`db`.`dim_table`
  FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id

Pushed-down filter support
- Existing source filters are reused when loading the cache.
- Join conditions such as:
```
ON o.user_id = u.user_id AND u.city = 'beijing'
```
  are applied together with the lookup key condition.
Full in-memory lookup cache
- The whole projected dimension table is loaded into a cache on the TaskManager heap, and every lookup is served from it, never falling back to the table.
- A dimension key may match multiple rows; all matching rows are returned and joined.
- Only lookup.cache=FULL is accepted; NONE and PARTIAL are rejected, because an Iceberg table cannot be point-looked-up effectively.
Configurable load strategy
- lookup.full-cache.eager-load selects when the cache is loaded:
  - false (default): on the first lookup. Simple, but that probe row is blocked for the duration of the load.
  - true: in open(), so no probe row is blocked, and a load that fails fails the job at startup instead of mid-stream.
Metrics
- Reported under the icebergLookupCache group: cacheHit and cacheMiss counters, and snapshotId and cachedRows gauges.

How to Use

Basic lookup join

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users` FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

Load the cache when the lookup function opens

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users`
/*+ OPTIONS('lookup.full-cache.eager-load' = 'true') */
FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

or

CREATE TABLE dim_users (
  user_id BIGINT,
  name STRING,
  city STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = 'iceberg_catalog',
  'catalog-type' = 'hadoop',
  'warehouse' = '/path/to/warehouse',
  'catalog-database' = 'db',
  'catalog-table' = 'users',

  'lookup.full-cache.eager-load' = 'true'
);

Options

Option	Required	Default	Description
`lookup.full-cache.eager-load`	No	`true`	Whether to load the cache in `open()`, instead of on the first lookup.

Both can be set per join with an OPTIONS hint, or in the table DDL WITH clause.

Notes

Memory only. The cache lives on the TaskManager heap, so this targets dimension tables that fit comfortably there. A disk-backed backend is left for a follow-up — keeping it out of this PR avoids adding native code to the runtime jar and any conflict with the RocksDB copy Flink already ships for its RocksDB state backend.
Loaded once, no refresh. The cache is loaded when the lookup function starts serving and is then kept for the lifetime of the job; there is no background reload. The dimension table should be populated before the join starts.

Guosmilesmile · 2026-09-16T16:48:52Z

For snapshot pinning, I was thinking we could resolve the snapshot at job submission time and pass the snapshot ID to every TaskManager, so that all caches use the same snapshot. But there is a problem with this approach.

The problem is failover. Flink doesn't re-run the planner when a task restarts, so the pinned snapshot ID would remain unchanged. If that snapshot has expired by then, the cache load would fail with Cannot find snapshot with ID ... after the restart, and retries would keep failing until the job is resubmitted.

Periodic refresh doesn't help here, because the failure happens during the initial cache load. We also don't have a way to persist and update the snapshot ID from the connector side, since a lookup join doesn't have checkpointed state.

So while pinning the snapshot would ensure consistency across subtasks, it introduces a failover problem.

Additionally, dimension tables do not change frequently.

For this PR, I therefore use the latest snapshot at the moment and just added logging and a metric to report the snapshot ID being used.If you have a better idea, we'd be happy to take a look.

swapna267 · 2026-09-17T00:12:00Z

why not use ReadOptions and pass when creating the IcebergFullCachingLookupFunction, which will keep it consistent across task restarts or job restarts.

At table creation time,

CREATE TABLE dim_users (
  user_id BIGINT,
  name STRING,
  city STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = 'iceberg_catalog',
  'catalog-type' = 'hadoop',
  'warehouse' = '/path/to/warehouse',
  'catalog-database' = 'db',
  'catalog-table' = 'users',

  'lookup.full-cache.eager-load' = 'true',
  'snapshot-id' = '121334'
);

OR per query

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users`
/*+ OPTIONS(
     'lookup.full-cache.eager-load' = 'true',
     'snapshot-id' = '121334'
   ) */
FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

There could be other strategies to simplify , instead of user specifying exact snapshot id also like latest or specific tag or as-of-timestamp

Guosmilesmile · 2026-09-17T01:32:54Z

@swapna267 Showing a specific snapshot ID is one approach, but it doesn't solve the problem of that snapshot expiring. On top of that, most users won't bother specifying a particular snapshot anyway.

swapna267 · 2026-09-17T04:23:15Z

Yes it doesn't need to be particular snapshot Id. Instead it could be Latest_Snapshot or a particular Branch/Tag. I was referring to resolving it on Job Submission time instead of on TaskManagers. But i just realized you already mentioned that option.

Instead of being inconsistent with lookup result across TaskManagers, prefer to have all TM's load from same Snapshot Id. And fail loudly, incase of an expired snapshot or a Tag.

Guosmilesmile · 2026-09-17T04:55:47Z

@swapna267 I agree that resolving the snapshot ID at Job Submission time works well when the cache is not refreshed, with an explicit failure if the snapshot has expired.

However, with periodic refresh, a long-running job may encounter a failover after the original snapshot has expired. Resolving the snapshot ID at Job Submission would require manual intervention to restart the job, which doesn't seem ideal for the periodic refresh use case.

I also noticed that other connectors supporting lookup joins generally establish their connections independently on each TaskManager.

So I don't think resolving the snapshot ID at the Job Submission level is a good fit for this use case.

This is also why I'm still leaning toward using the latest snapshot rather than pinning to a specific snapshot.

Guosmilesmile · 2026-09-18T10:52:18Z

@pvary @mxm @talatuyarer If you get a chance, please take a look and let me know what you think. I'd really appreciate it.

swapna267 · 2026-09-18T15:44:57Z

Thanks @Guosmilesmile . Yes I agree, this wouldn't make sense for Periodic Refresh .
With periodic refresh, falling back to Latest makes sense.

As this PR's scope was limited to one time load with no periodic refresh, i was recommending that.

pvary · 2026-09-22T08:21:45Z

+| Option                       | Default | Description                                                                                                                 |
+| ---------------------------- |---------|-----------------------------------------------------------------------------------------------------------------------------|
+| lookup.cache                 |         | Only `FULL` is accepted; `NONE` and `PARTIAL` are rejected, because an Iceberg table cannot be point-looked-up effectively. |
+| lookup.full-cache.eager-load | true    | Whether to load the full cache when the lookup function is opened, instead of on the first lookup.                          |


we usually do hiearchical. So maybe lookup.cache.full.eager-load, or since we don't have full, we can just say lookup.cache.eager-load?

I used the lookup.full-cache. prefix because I’d like the refresh-related options we add later to be consistent with the existing Flink options.

https://github.com/apache/flink/blob/master/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/lookup/LookupOptions.java

Makes sense

pvary · 2026-09-22T08:31:36Z

+import org.apache.flink.table.data.RowData;
+
+@Internal
+interface IcebergLookupCache extends Closeable {


Shall we introduce this only once we have more cache implementations?

Make sense, at first I think we will add other back end in the future. Remove it now.

pvary · 2026-09-22T08:33:55Z

+
+  @Override
+  public void add(RowData key, RowData row) {
+    cache.computeIfAbsent(key, k -> Lists.newLinkedList()).add(row);


nit: why linked list?

This was my mistake. I misjudged the scale of the data here.

pvary · 2026-09-22T08:34:34Z

+    RowType projectedRowType = (RowType) projected.toPhysicalRowDataType().getLogicalType();
+    List<Expression> pushedFilters = filters == null ? ImmutableList.of() : filters;
+
+    Configuration lookupConf = Configuration.fromMap(properties);


Why not FlinkConfParser?

Changed it to consistently use FlinkConfParser.

pvary · 2026-09-22T08:52:21Z

+    lookupFunction = newLookupFunction();
+    lookupFunction.open(new FunctionContext(null));
+
+    Collection<RowData> rows = lookupFunction.lookup(keyRow(1L));


add test with multicolumn key

Add it now.

pvary · 2026-09-22T08:55:45Z

+    assertThat(rows).singleElement().satisfies(row -> assertRow(row, 1L, "alice", "A"));
+
+    // Served from the cache, not re-read from the table.
+    assertThat(lookupFunction.lookup(keyRow(1L))).isSameAs(rows);


I don't think we should rely on the cache returning the exact rows is a good idea. We might change the implementation which could return a copy of the rows, or something.

Could we just assert on the contents, or on some metrics?

Yes, add a method assert on the contents.

pvary · 2026-09-23T08:05:33Z

+
+| Option                       | Default | Description                                                                                                                 |
+| ---------------------------- |---------|-----------------------------------------------------------------------------------------------------------------------------|
+| lookup.full-cache.eager-load | true    | Whether to load the full cache when the lookup function is opened, instead of on the first lookup.                          |


nit: use `-s, and format the table

Also we could add it to docs/docs/flink-configuration.md,?

Add it now.

pvary · 2026-09-23T08:07:32Z

+
+Iceberg implements lookup join with a full cache: the whole projected dimension table is loaded into the cache, and every lookup is served from it without falling back to the table. The full cache is held in memory on the TaskManager heap, so lookup join targets dimension tables that fit comfortably there.
+
+The cache is loaded by default when the lookup function is opened. Set lookup.full-cache.eager-load to false to load it on the first lookup instead, which blocks the data flow until the cache is fully loaded.


Reword to make it clear that this is a decision between where/when to block

pvary · 2026-09-23T08:08:15Z

+ * A full caching lookup function: the whole projected Iceberg dimension table is loaded into an
+ * in-memory cache on the first lookup, and every lookup is served from that cache.
+ */
+public class IcebergFullCachingLookupFunction extends LookupFunction {


do we need this to be public?

Yes, we call this class in other package, I add Internal for this class now.

pvary · 2026-09-23T08:09:08Z

+  private transient Counter cacheHitCounter;
+  private transient Counter cacheMissCounter;


Do we need these with only full cache supported?

Kept only lookupMiss (probe rows whose key is not in the cached dimension table) and dropped cacheHit, which for a full cache simply equals the number of lookups.

Do we have stat for the number of lookups?

pvary · 2026-09-23T08:16:12Z

+    this.flinkConfParser = new FlinkConfParser(properties, readableConfig);
+    this.caseSensitive =
+        flinkConfParser
+            .booleanConf()
+            .option(FlinkReadOptions.CASE_SENSITIVE)
+            .flinkConfig(FlinkReadOptions.CASE_SENSITIVE_OPTION)
+            .defaultValue(FlinkReadOptions.CASE_SENSITIVE_OPTION.defaultValue())
+            .parse();


Maybe put into the getLookupRuntimeProvider?

pvary · 2026-09-23T08:17:55Z

+
+The cache is loaded by default when the lookup function is opened. Set lookup.full-cache.eager-load to false to load it on the first lookup instead, which blocks the data flow until the cache is fully loaded.
+
+There is no background refresh: the cache keeps the data it was loaded with for the lifetime of the job, so the dimension table should be populated before the join starts.


Mention that the cache could be out of sync between the tasks

Fix it now.

pvary · 2026-09-23T08:18:34Z

    options.add(FlinkCreateTableOptions.CATALOG_TABLE);
    options.add(FlinkCreateTableOptions.USE_DYNAMIC_ICEBERG_SINK);
    options.add(FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL);
+    options.add(LookupOptions.CACHE_TYPE);


Do we want to remove this?

Not strictly needed for behavior, but I kept it as a declaration. optionalOptions() is the only place that lists the supported options, and once the factory starts using TableFactoryHelper.validate(), an undeclared lookup.cache would fail. Happy to drop it if you'd rather.

Remove it if we don't use it now. We can always add back later

pvary · 2026-09-23T10:03:14Z

+
+    LOG.info(
+        "IcebergFullCachingLookupFunction loading started, snapshot={}, committedAt={}, projected fields={}, pushedFilters={}",
+        snapshotId == IcebergLookupReader.CURRENT_SNAPSHOT ? "none" : snapshotId,


shall we use null instead of a constant?

pvary · 2026-09-23T10:04:01Z

+    Schema tableSchema = table.schema();
+    List<String> projectedColumns = projectedRowType.getFieldNames();
+    Types.NestedField[] projectedFields = new Types.NestedField[projectedColumns.size()];
+    for (int i = 0; i < projectedColumns.size(); i++) {
+      String column = projectedColumns.get(i);
+      Types.NestedField field = tableSchema.findField(column);
+      Preconditions.checkArgument(field != null, "Cannot find column '%s' in table schema", column);
+      projectedFields[i] = field;
+    }
+
+    Schema icebergProjection = new Schema(projectedFields);


Suggested change

Schema tableSchema = table.schema();

List<String> projectedColumns = projectedRowType.getFieldNames();

Types.NestedField[] projectedFields = new Types.NestedField[projectedColumns.size()];

for (int i = 0; i < projectedColumns.size(); i++) {

String column = projectedColumns.get(i);

Types.NestedField field = tableSchema.findField(column);

Preconditions.checkArgument(field != null, "Cannot find column '%s' in table schema", column);

projectedFields[i] = field;

}

Schema icebergProjection = new Schema(projectedFields);

Schema icebergProjection =

FlinkSchemaUtil.convert(table.schema(), FlinkSchemaUtil.toResolvedSchema(projectedRowType));

pvary · 2026-09-23T10:14:11Z

+    assertThat(lookupFunction.lookup(keyRow(null, "A")))
+        .singleElement()
+        .satisfies(row -> assertRow(row, null, "nobody", "A"));


Is this required from the Flink side?

How can we reach this with SQL?

pvary · 2026-09-23T10:17:38Z

+import org.junit.jupiter.api.Test;
+
+/** Tests SQL lookup join with Iceberg table source. */
+class TestIcebergLookupJoinSql extends TestSqlBase {


shall we add an eager load test?

pvary · 2026-09-23T10:19:37Z

+options in the DDL, or per query with the `OPTIONS` hint:
+
+```sql
+SELECT o.order_id, u.name


SET table.dynamic-table-options.enabled=true;

Maybe remove the example from here and just link the other page?

pvary · 2026-09-23T10:26:06Z

+The lookup options are:
+
+| Option                         | Default | Description                                                                                        |
+| ------------------------------ | ------- | -------------------------------------------------------------------------------------------------- |
+| `lookup.full-cache.eager-load` | `true`  | Whether to load the full cache when the lookup function is opened, instead of on the first lookup. |
+


Maybe just link the config here.

The main goal is to write everything only once

github-actions Bot added flink docs labels Sep 16, 2026

Guosmilesmile force-pushed the lookup_join_heap branch from 6550de0 to de77ef7 Compare September 16, 2026 16:56

Guosmilesmile mentioned this pull request Sep 16, 2026

Flink: Support Lookup join #17280

Closed

Guosmilesmile force-pushed the lookup_join_heap branch from de77ef7 to 81931fe Compare September 17, 2026 06:36

Guosmilesmile mentioned this pull request Sep 17, 2026

Flink: Add periodic and timed reload of the lookup join full cache #18149

Open

Guosmilesmile force-pushed the lookup_join_heap branch from 81931fe to 41914e4 Compare September 17, 2026 10:18